package org.example;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * 自定义TableFunction，用于将JSON数组展开为多行数据
 * 专门用于处理CDC消息中的数据数组字段，如$.data
 * 
 * 此函数将单个JSON数组字符串转换为多行输出，每行包含:
 * - data_idx: 元素在数组中的索引位置
 * - data_element: 元素的JSON字符串表示
 * 
 * 使用场景：
 * 1. 处理CDC数据中包含多行变更的消息
 * 2. 展开JSON数组为表格行，便于后续处理
 * 3. 配合LATERAL TABLE用于SQL中的数组展开操作
 * 
 * 示例：
 * 假设输入JSON数组字符串为:
 * ```
 * [
 *   {"id": "1", "name": "张三", "age": 25},
 *   {"id": "2", "name": "李四", "age": 30}
 * ]
 * ```
 * 
 * 使用此函数后，会生成2行数据:
 * | data_idx | data_element                        |
 * |----------|-------------------------------------|
 * | 0        | {"id": "1", "name": "张三", "age": 25} |
 * | 1        | {"id": "2", "name": "李四", "age": 30} |
 * 
 * 在Flink SQL中的使用示例:
 * ```sql
 * -- 注册函数
 * CREATE TEMPORARY FUNCTION JSON_ARRAY_TABLE AS 'org.example.JsonArrayFunction';
 * 
 * -- 应用函数展开数组
 * SELECT t.data_idx, j.* 
 * FROM source_table,
 * LATERAL TABLE(JSON_ARRAY_TABLE(source_table.json_array_field)) AS t(data_idx, data_element),
 * JSON_TO_MAP(t.data_element) AS j
 * ```
 */
@FunctionHint(
    output = @DataTypeHint("ROW<data_idx INT, data_element STRING>")
)
public class JsonToArrayFunction extends TableFunction<Row> {
    
    private static final ObjectMapper objectMapper = new ObjectMapper();
    
    /**
     * 将JSON数组字符串展开为多行
     * 
     * 此方法处理不同的输入情况:
     * 1. 如果输入为null或空字符串，不输出任何行
     * 2. 如果输入不是JSON数组，将其作为索引为0的单个元素处理
     * 3. 如果输入是JSON数组，则为每个数组元素生成一行，索引从0开始
     * 
     * @param jsonArrayStr JSON数组字符串，例如: "[{\"id\":\"1\"},{\"id\":\"2\"}]"
     *                    或单个JSON对象: "{\"id\":\"1\"}"
     */
    public void eval(String jsonArrayStr) {
        if (jsonArrayStr == null || jsonArrayStr.isEmpty()) {
            return;
        }
        
        try {
            // 解析JSON数组
            JsonNode jsonNode = objectMapper.readTree(jsonArrayStr);
            
            // 如果不是数组，作为单个元素处理
            if (!jsonNode.isArray()) {
                // 创建包含索引和JSON内容的行
                Row row = Row.of(0, jsonNode.toString());
                collect(row);
                return;
            }
            
            // 处理数组节点
            ArrayNode arrayNode = (ArrayNode) jsonNode;
            for (int i = 0; i < arrayNode.size(); i++) {
                JsonNode element = arrayNode.get(i);
                // 创建包含索引和JSON内容的行
                Row row = Row.of(i, element.toString());
                collect(row);
            }
        } catch (Exception e) {
            System.out.println("JSON数组解析异常: " + e.getMessage() + ", JSON: " + jsonArrayStr);
        }
    }
} 